package com.yeeiee.convert.impl;

import cn.hutool.db.Entity;
import com.yeeiee.convert.ClinkConverter;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import lombok.var;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.catalog.exceptions.CatalogException;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.*;

import java.util.Locale;

@Slf4j
public class MysqlConverter implements ClinkConverter {


    @Override
    public void convertToType(Entity entity, Schema.Builder builder) {
        val columnName = entity.getStr("column_name");
        val sourceType = entity.getStr("source_type");
        var typeName = entity.getStr("type_name").toUpperCase();
        if (sourceType.toLowerCase(Locale.ROOT).contains("unsigned")) {
            typeName += "_UNSIGNED";
        }
        val precision = entity.getInt("precision");
        val scale = entity.getInt("scale");
        val isNullable = entity.getBool("is_nullable");
        val columnCharacterLength = entity.getInt("column_character_length");
        val comment = entity.getStr("comment");

        DataType dataType;

        switch (typeName) {
            case "VARCHAR":
            case "TINYTEXT":
            case "MEDIUMTEXT":
            case "TEXT":
            case "ENUM":
            case "SET":
                dataType = DataTypes.of(new VarCharType(isNullable, columnCharacterLength));
                break;
            case "TINYINT":
                if ("tinyint(1)".equals(sourceType)) {
                    dataType = DataTypes.of(new BooleanType(isNullable));
                } else {
                    dataType = DataTypes.of(new IntType(isNullable));
                }
                break;
            case "TINYINT_UNSIGNED":
            case "SMALLINT":
                dataType = DataTypes.of(new SmallIntType(isNullable));
                break;
            case "SMALLINT_UNSIGNED":
            case "MEDIUMINT":
            case "MEDIUMINT_UNSIGNED":
            case "INT":
                dataType = DataTypes.of(new IntType(isNullable));
                break;
            case "INT_UNSIGNED":
            case "BIGINT":
                dataType = DataTypes.of(new BigIntType(isNullable));
                break;
            case "BIGINT_UNSIGNED":
                dataType = DataTypes.of(new DecimalType(isNullable, 20, 0));
                break;
            case "DECIMAL":
                dataType = DataTypes.of(new DecimalType(isNullable, precision, scale));
                break;
            case "DECIMAL_UNSIGNED":
                dataType = DataTypes.of(new DecimalType(isNullable, precision + 1, scale));
                break;
            case "FLOAT":
            case "FLOAT_UNSIGNED":
                dataType = DataTypes.of(new FloatType(isNullable));
                break;
            case "DOUBLE":
            case "DOUBLE_UNSIGNED":
                dataType = DataTypes.of(new DoubleType(isNullable));
                break;
            case "CHAR":
                dataType = DataTypes.of(new CharType(isNullable, columnCharacterLength));
                break;
            case "JSON":
            case "LONGTEXT":
                dataType = DataTypes.of(new VarCharType(isNullable, VarCharType.MAX_LENGTH));
                break;
            case "DATE":
                dataType = DataTypes.of(new DateType(isNullable));
                break;
            case "TIME":
                dataType = DataTypes.of(new TimeType(isNullable, TimeType.DEFAULT_PRECISION));
                break;
            case "DATETIME":
            case "TIMESTAMP":
                dataType = DataTypes.of(new TimestampType(isNullable, TimestampType.DEFAULT_PRECISION));
                break;
            case "BIT":
            case "TINYBLOB":
            case "MEDIUMBLOB":
            case "BLOB":
            case "LONGBLOB":
            case "VARBINARY":
            case "BINARY":
                dataType = DataTypes.of(new VarBinaryType(isNullable, VarBinaryType.MAX_LENGTH));
                break;
            default:
                throw new CatalogException("Unsupported data type: " + typeName);
        }

        builder.column(columnName, dataType).withComment(comment);
    }

    @Override
    public String convertToDb(Schema.UnresolvedColumn column) {
        if (!(column instanceof Schema.UnresolvedPhysicalColumn)) {
            throw new CatalogException("Unsupported column type: " + column);
        }
        val physicalColumn = (Schema.UnresolvedPhysicalColumn) column;
        val dataType = (DataType) physicalColumn.getDataType();

        val logicalType = dataType.getLogicalType();

        String type = null;
        if (logicalType instanceof VarCharType) {
            val length = ((VarCharType) logicalType).getLength();
            // 假定utf8mb4
            val threshold = 65532 / 4;
            if (length > threshold) {
                type = "TEXT";
            } else {
                type = String.format("VARCHAR(%s)", length);
            }
        } else if (logicalType instanceof BooleanType) {
            type = "BOOLEAN";
        } else if (logicalType instanceof SmallIntType) {
            type = "SMALLINT";
        } else if (logicalType instanceof IntType) {
            type = "INT";
        } else if (logicalType instanceof BigIntType) {
            type = "BIGINT";
        } else if (logicalType instanceof DecimalType) {
            val decimalType = (DecimalType) logicalType;
            type = String.format("DECIMAL(%s,%s)", decimalType.getPrecision(), decimalType.getScale());
        } else if (logicalType instanceof FloatType) {
            type = "FLOAT";
        } else if (logicalType instanceof DoubleType) {
            type = "DOUBLE";
        } else if (logicalType instanceof CharType) {
            val length = ((CharType) logicalType).getLength();
            type = String.format("CHAR(%s)", length);
        } else if (logicalType instanceof DateType) {
            type = "DATE";
        } else if (logicalType instanceof TimeType) {
            type = "TIME";
        } else if (logicalType instanceof TimestampType) {
            type = "DATETIME";
        } else if (logicalType instanceof VarBinaryType) {
            val length = ((VarBinaryType) logicalType).getLength();
            val threshold = 65532;
            if (length > threshold) {
                type = "BLOB";
            } else {
                type = String.format("VARBINARY(%s)", length);
            }
        } else if (logicalType instanceof BinaryType) {
            val length = ((BinaryType) logicalType).getLength();
            val threshold = 65532;
            if (length > threshold) {
                type = "BLOB";
            } else {
                type = String.format("BINARY(%s)", length);
            }
        }

        if (type == null) {
            throw new CatalogException("Unsupported type: " + logicalType);
        }

        if (!logicalType.isNullable()) {
            type = type + "  NOT NULL";
        }

        return type;
    }
}
